package com.atguigu.cn.dataStream.sink

import com.atguigu.cn.dataStream.api.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

/**
  * @author: yangShen
  * @Description:
  * @Date: 2020/4/15 9:06 
  */
object RedisSinkTest {
  def main(args: Array[String]): Unit = {

    //1.environment
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    // 2.source
    val inputStream = environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\flink-tutorial\\src\\main\\resources\\sensor.txt")

    // 3.transform 操作，但没有标记返回类型
    val dataStream = inputStream.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })

    val conf = new FlinkJedisPoolConfig.Builder().setHost("10.16.26.16").setPort(6379).build()

    //sink -- 往redis 中写数据
    //命令验证查看：1:keys *; 2:hget sensor_temperature;
    dataStream.addSink(new RedisSink(conf, new MyRedisMapper()))

    environment.execute("redis sink test")

  }

}

class MyRedisMapper() extends RedisMapper[SensorReading] {
  // 定义保存数据到redis的命令
  override def getCommandDescription: RedisCommandDescription = {
     //应用场景：一个传感器对应一个温度值，传感器实时变化也把温度值实时更新，另外一个程序实时获取
    //把传感器id和温度值保存成哈希表，HEST key(表的名)      field value (一对key,value)
    new RedisCommandDescription(RedisCommand.HSET, "sensor_temperature")  //sensor_temperature:表名
  }

  //定义保存到redis的key
  override def getKeyFromData(t: SensorReading): String = t.id

  //定义保存到redis的value
  override def getValueFromData(t: SensorReading): String = t.temperature.toString
}
